package com.digiwin.demo.influx;

import com.digiwin.demo.vo.VisitVO;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.influxdb.dto.Point;

import java.util.concurrent.TimeUnit;

/**
 * @Author yanggld
 * @Date 2020/1/2-16:45
 * @Description
 */
public class InfluxSinkFunction extends RichSinkFunction<VisitVO> {

	private InfluxTemplate template;

	@Override
	public void open(Configuration parameters) throws Exception {
		InfluxBean bean= InfluxBean.builder().dbName("test")
				.url("http://172.16.1.198:9999")
				.username("root")
				.password("12345678")
				.build();
		template = new SimpleInfluxTemplate(bean);
	}

	@Override
	public void close() throws Exception {
		template.close();
	}

	@Override
	public void invoke(VisitVO value, Context context) throws Exception {
		template.write(Point.measurement("sport")
				.addField("count",value.getCount())
				.tag("uid",value.getUid())
				.tag("name",value.getName())
				.time(context.currentProcessingTime(), TimeUnit.MILLISECONDS).build());
	}
}
